package dkbta.processing.data;

import java.io.BufferedWriter;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import dkbta.ontology.elements.Element;
import dkbta.ontology.elements.ElementDef;
import dkbta.processing.ActionTimer;
import dkbta.util.Log;
import dkbta.util.time.TimeInterval;

/**
 * A thread that is in charge of sending data (element instances) to other units and to
 * the DB. It operates on a producer-consumer basis (it is the consumer and the producer
 * is the work queue). <br>
 * <br>
 * Note this is a non-daemon thread, i.e. the unit will not shut down until all
 * transmissions have finished
 * 
 * @author Uri Kanonov
 */
public class DataDispatcher extends Thread{
	/** The data batches that are waiting to be sent */
	private final BlockingQueue<DataBatch> _outgoingData;

	/** Database information */
	private DB _db;

	/** Whether the dispatcher should shut down */
	private boolean _shouldFinish;

	/** A log handle */
	private final Log _log;

	/**
	 * Initializes (however doesn't start) the data dispatcher
	 */
	public DataDispatcher(){
		super("Data dispatcher");
		_outgoingData = new LinkedBlockingQueue<DataBatch>();
		_shouldFinish = false;
		_log = Log.getLogger();
		/*
		 * CHANGE-1 Arkady Mishiev 27.05.01
		 */
		setDaemon(true);
	}

	/**
	 * Adds a data batch to be sent
	 * 
	 * @param dataBatch The data batch to be sent
	 */
	public void addDataBatch(DataBatch dataBatch){
		_outgoingData.add(dataBatch);
	}

	@Override
	public void run(){
		Log log = Log.getLogger();
		ActionTimer actionTimer = ActionTimer.getInstance();
		try{
			while (!shouldFinish()){
				DataBatch db = _outgoingData.take();
				SendTo sendTo = db.sendTo;
				int pid = db.patientId;
				Collection<InetSocketAddress> units = sendTo.units;
				ElementDef elementDef = db.elementDef;

				if (!(units == null || units.isEmpty())){
					int dataSendingActionId = actionTimer.actionStart();
					String data = db.serialize();

					// Opening a connection to each of the destinations and sending them
					// the serialized contents
					for (InetSocketAddress unit : units){
						_log.finest("Started sending " + db + " to unit: " + unit);

						try{
							Socket socket = new Socket();
							socket.connect(unit);
							BufferedWriter wr = new BufferedWriter(
									new OutputStreamWriter(socket.getOutputStream()));
							wr.write(data);
							wr.flush();

							// Waiting until the other side finishes reading
							// and closes the socket
							socket.getInputStream().read();
						}catch(Exception e){
							log.warning("Unable to send " + db + " to: " + unit, e);
						}
						_log.finest("Finished sending " + db + " to unit: " + unit);
					}
					actionTimer.actionEnd(dataSendingActionId, pid, elementDef.getName(),
						elementDef.getElementType(),
						ActionTimer.Action.SENDING_DATA_TO_UNITS);
				}

				try{
					PreparedStatement sendDataPS;
					if (sendTo.sendToDB && _db != null
							&& (sendDataPS = _db.getSendDataPS()) != null){
						int dataSendingActionId = actionTimer.actionStart();
						List<? extends Element> elements = db.elements;
						// SubjectID, ConceptName, StartTime, EndTime, ConceptValue,
						// ConceptType
						for (Element e : elements){
							TimeInterval ti = e.getTimeInterval();
							sendDataPS.setInt(1, pid);
							sendDataPS.setString(2, e.getName());
							sendDataPS.setTimestamp(3, new Timestamp(ti.getStartTime()));
							sendDataPS.setTimestamp(4, new Timestamp(ti.getEndTime()));
							e.insertDataIntoPS(sendDataPS, 5, 6);
							sendDataPS.executeUpdate();
						}
						actionTimer.actionEnd(dataSendingActionId, pid, elementDef
								.getName(), elementDef.getElementType(),
							ActionTimer.Action.SENDING_DATA_TO_DB);
					}
				}catch(SQLException e){
					log.warning("Unable to send " + db + " to DB", e);
				}
			}
		}catch(InterruptedException e){}catch(Throwable t){
			log.severe("Data dispatcher thread died", t);
		}
//		log.severe("DataDispatcher finished");
	}

	/**
	 * Sets the db info
	 * 
	 * @param db The db info
	 */
	public void setDB(DB db){
		_db = db;
	}

	/**
	 * Tests whether the data dispatcher is done. It's done if signaled so by the work
	 * queue (there will be no future tasks) and that the outgoing data queue is empty
	 * (there are no pending tasks).
	 * 
	 * @return Whether the data dispatcher is done and should shut down
	 */
	private synchronized boolean shouldFinish(){
		return _shouldFinish && _outgoingData.isEmpty();
	}

	/**
	 * Telling the data dispatcher there won't be any further data batches coming and that
	 * it can shut down as soon as it finishes sending the ones it has. This method will
	 * return once the data dispatcher has finished sending all of its data (and
	 * registering the sending times)
	 * 
	 * @throws InterruptedException If the waiting thread is interrupted before the data
	 *         dispatcher finishes
	 */
	public synchronized void waitForFinish() throws InterruptedException{
		_shouldFinish = true;
		if (_outgoingData.isEmpty()){
			interrupt();
		}
		join();
	}
}
